package com.xian.scala.stream

import java.util.Properties

import org.apache.flink.api.common.serialization.{DeserializationSchema, SimpleStringSchema}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

/**
 * 读取(消费)kafka信息
 */
object ReadKafkaData {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //kafka信息
    //没有就创建
    // kafka-topics.sh --create --bootstrap-server 172.18.4.230:9092,172.18.4.231:9092,172.18.4.232:9092 --replication-factor 3 --partitions 4 --topic jmy-topic
    val topic = "jmy-topic"
    val props = new Properties()
    props.setProperty("bootstrap.servers","172.18.4.230:9092,172.18.4.231:9092,172.18.4.232:9092")
    props.setProperty("group.id", "jemson-consumer-group")
    props.setProperty("key.deserializer",
      "org.apache.kafka.common.serialization.StringDeserializer")
    props.setProperty("value.deserializer",
      "org.apache.kafka.common.serialization.StringDeserializer")
    props.setProperty("auto.offset.reset", "latest")

    val ds = env.addSource(new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), props))

    ds.print()


    env.execute()



  }
}
